package com.example.transformation;

import com.example.bean.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created with IntelliJ IDEA.
 * ClassName: MapDemo
 * Package: com.example.transformation
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-18
 * Time: 16:03
 */

public class FiatMapDemo {
    public static void main(String[] args) throws Exception {
        //1.创建环境
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);


        /**
         * flatMap 可以一进多出 一出 不出
         * map和 flatMap的区别
         * map方法使用的return 必须是要返回一个东西的
         * flatMap方法没有返回值 输出数据是考采集器 调用几次输出几个
         */

        //fromElements 是直接赋值数据
        DataStreamSource<WaterSensor> data = env.fromElements(
                //赋值WaterSensor对象的数据
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s1", 11L, 11),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s3", 3L, 3)
        );
        //如果输入的数据是sensor_1，只打印vc；如果输入的数据是sensor_2，既打印ts又打印vc。

        final SingleOutputStreamOperator<String> flatMap = data.flatMap(new FlatMapFunction<WaterSensor, String>() {
            @Override
            public void flatMap(WaterSensor waterSensor, Collector<String> collector) throws Exception {
                if ("s1".equals(waterSensor.getId())) {
                    //Collector 采集器 收集器 想下游发送数据
                    collector.collect(waterSensor.getVc().toString());
                } else if ("s2".equals(waterSensor.getId())) {
                    //分别输出 就是发送两次
                    collector.collect(waterSensor.getTs().toString());
                    collector.collect(waterSensor.getVc().toString());
                } else {

                }
            }
        });
        flatMap.print();
        env.execute();
    }

}
